package printnum;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

import java.util.Map;
import java.util.Random;

/**
 * 创建一个数据源组件，产生一些模拟数据
 */
public class NumberSpout extends BaseRichSpout {

    private SpoutOutputCollector collector;

    /**
     * init method, 用于组件的初始化
     */
    @Override
    public void open(Map<String, Object> map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        collector = spoutOutputCollector;
    }

    /**
     * 用于产生数据，并向下游发送Tuple
     */
    @Override
    public void nextTuple() {
        // 产生一个100以内的随机数
        int number = new Random().nextInt(100);

        // 将数字封装到Tuple中，然后通过collector将Tuple发射出去
        collector.emit(new Values(number));

        // 一定不要加阻塞方法，此处纯粹是为了测试
        Utils.sleep(1000);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        // 通过Fields声明Tuple的字段名称
        outputFieldsDeclarer.declare(new Fields("number"));
    }
}
